/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.nageoffer.shortlink.project.mq.consumer;

import cn.hutool.core.date.DateUtil;
import cn.hutool.core.date.Week;
import cn.hutool.core.util.StrUtil;
import cn.hutool.http.HttpUtil;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.nageoffer.shortlink.project.common.convention.exception.ServiceException;
import com.nageoffer.shortlink.project.dao.entity.LinkAccessLogsDO;
import com.nageoffer.shortlink.project.dao.entity.LinkAccessStatsDO;
import com.nageoffer.shortlink.project.dao.entity.LinkBrowserStatsDO;
import com.nageoffer.shortlink.project.dao.entity.LinkDeviceStatsDO;
import com.nageoffer.shortlink.project.dao.entity.LinkLocaleStatsDO;
import com.nageoffer.shortlink.project.dao.entity.LinkNetworkStatsDO;
import com.nageoffer.shortlink.project.dao.entity.LinkOsStatsDO;
import com.nageoffer.shortlink.project.dao.entity.LinkStatsTodayDO;
import com.nageoffer.shortlink.project.dao.entity.ShortLinkGotoDO;
import com.nageoffer.shortlink.project.dao.mapper.LinkAccessLogsMapper;
import com.nageoffer.shortlink.project.dao.mapper.LinkAccessStatsMapper;
import com.nageoffer.shortlink.project.dao.mapper.LinkBrowserStatsMapper;
import com.nageoffer.shortlink.project.dao.mapper.LinkDeviceStatsMapper;
import com.nageoffer.shortlink.project.dao.mapper.LinkLocaleStatsMapper;
import com.nageoffer.shortlink.project.dao.mapper.LinkNetworkStatsMapper;
import com.nageoffer.shortlink.project.dao.mapper.LinkOsStatsMapper;
import com.nageoffer.shortlink.project.dao.mapper.LinkStatsTodayMapper;
import com.nageoffer.shortlink.project.dao.mapper.ShortLinkGotoMapper;
import com.nageoffer.shortlink.project.dao.mapper.ShortLinkMapper;
import com.nageoffer.shortlink.project.dto.biz.ShortLinkStatsRecordDTO;
import com.nageoffer.shortlink.project.mq.idempotent.MessageQueueIdempotentHandler;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.redisson.api.RLock;
import org.redisson.api.RReadWriteLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import static com.nageoffer.shortlink.project.common.constant.RedisKeyConstant.LOCK_GID_UPDATE_KEY;
import static com.nageoffer.shortlink.project.common.constant.ShortLinkConstant.AMAP_REMOTE_URL;

/**
 * 短链接监控状态保存消息队列消费者
 */
@Slf4j
@Component
@RequiredArgsConstructor
@RocketMQMessageListener(
        topic = "${rocketmq.producer.topic}",
        consumerGroup = "${rocketmq.consumer.group}"
)
public class ShortLinkStatsSaveConsumer implements RocketMQListener<Map<String, String>> {

    private final ShortLinkMapper shortLinkMapper;
    private final ShortLinkGotoMapper shortLinkGotoMapper;
    private final RedissonClient redissonClient;
    private final LinkAccessStatsMapper linkAccessStatsMapper;
    private final LinkLocaleStatsMapper linkLocaleStatsMapper;
    private final LinkOsStatsMapper linkOsStatsMapper;
    private final LinkBrowserStatsMapper linkBrowserStatsMapper;
    private final LinkAccessLogsMapper linkAccessLogsMapper;
    private final LinkDeviceStatsMapper linkDeviceStatsMapper;
    private final LinkNetworkStatsMapper linkNetworkStatsMapper;
    private final LinkStatsTodayMapper linkStatsTodayMapper;
    private final MessageQueueIdempotentHandler messageQueueIdempotentHandler;

    @Value("${short-link.stats.locale.amap-key}")
    private String statsLocaleAmapKey;

    @Override
    public void onMessage(Map<String, String> producerMap) {
        String keys = producerMap.get("keys");
        if (!messageQueueIdempotentHandler.isMessageProcessed(keys)) {
            // 判断当前的这个消息流程是否执行完成
            if (messageQueueIdempotentHandler.isAccomplish(keys)) {
                log.info("消息已处理完成，不再重复处理。keys: {}", keys);
                return;
            }
            throw new ServiceException("消息未完成流程，需要消息队列重试");
        }
        try {
            ShortLinkStatsRecordDTO statsRecord = JSON.parseObject(producerMap.get("statsRecord"), ShortLinkStatsRecordDTO.class);
            actualSaveShortLinkStats(statsRecord);
            messageQueueIdempotentHandler.setAccomplish(keys);
        } catch (Throwable ex) {
            log.error("记录短链接监控消费异常，keys: {}, producerMap: {}", keys, producerMap, ex);
            try {
                messageQueueIdempotentHandler.delMessageProcessed(keys);
            } catch (Throwable remoteEx) {
                log.error("删除幂等标识错误", remoteEx);
            }
            throw ex;
        }
    }

    /**
     * 实际保存短链接统计数据的核心方法
     *
     * @param statsRecord  统计记录数据传输对象
     */
    public void actualSaveShortLinkStats(ShortLinkStatsRecordDTO statsRecord) {
        // 1. 【参数预处理】确保fullShortUrl不为空，优先使用传入参数，否则从statsRecord中获取
        String fullShortUrl = statsRecord.getFullShortUrl();

        // 2. 【分布式锁机制】获取基于短链接URL的读写锁，防止并发更新时的数据不一致
        RReadWriteLock readWriteLock = redissonClient.getReadWriteLock(String.format(LOCK_GID_UPDATE_KEY, fullShortUrl));
        RLock rLock = readWriteLock.readLock();

        // 3. 【并发控制】尝试获取读锁，如果获取失败说明有写操作在进行，阻塞等待
        rLock.lock();
        try {
            // 4. 通过短链接URL查询数据库获取对应的分组ID
            LambdaQueryWrapper<ShortLinkGotoDO> queryWrapper = Wrappers.lambdaQuery(ShortLinkGotoDO.class)
                    .eq(ShortLinkGotoDO::getFullShortUrl, fullShortUrl);
            ShortLinkGotoDO shortLinkGotoDO = shortLinkGotoMapper.selectOne(queryWrapper);
            String gid = shortLinkGotoDO.getGid();

            // 5. 【时间维度数据准备】获取当前访问的小时和星期几，用于时间维度统计分析
            int hour = DateUtil.hour(new Date(), true);  // 获取24小时制的小时数
            Week week = DateUtil.dayOfWeekEnum(new Date());  // 获取星期枚举
            int weekValue = week.getIso8601Value();  // 转换为ISO8601标准的星期值(1-7)

            // 6. 【访问统计数据构建】创建链接访问统计对象，记录PV、UV、UIP等核心访问指标
            LinkAccessStatsDO linkAccessStatsDO = LinkAccessStatsDO.builder()
                    .pv(1)  // 页面访问量+1
                    .uv(statsRecord.getUvFirstFlag() ? 1 : 0)  // 如果是用户首次访问，UV+1
                    .uip(statsRecord.getUipFirstFlag() ? 1 : 0)  // 如果是IP首次访问，UIP+1
                    .hour(hour)  // 访问小时
                    .weekday(weekValue)  // 访问星期
                    .fullShortUrl(fullShortUrl)  // 完整短链接
                    .date(new Date())  // 访问日期
                    .build();
            // 7. 【保存访问统计】将访问统计数据持久化到数据库
            linkAccessStatsMapper.linkAccessStats(linkAccessStatsDO);

            // 8. 【地理位置信息获取】通过高德地图API根据访问者IP地址获取地理位置信息
            Map<String, Object> localeParamMap = new HashMap<>();
            localeParamMap.put("key", statsLocaleAmapKey);  // 高德地图API密钥
            localeParamMap.put("ip", statsRecord.getRemoteAddr());  // 访问者IP地址
            String localeResultStr = HttpUtil.get(AMAP_REMOTE_URL, localeParamMap);  // 调用高德API
            JSONObject localeResultObj = JSON.parseObject(localeResultStr);  // 解析API响应

            // 9. 【地理位置数据解析】解析高德API返回的地理位置信息
            String infoCode = localeResultObj.getString("infocode");  // 获取API调用状态码
            String actualProvince = "未知";  // 默认省份
            String actualCity = "未知";  // 默认城市

            // 10. 【地理位置统计】当API调用成功时，解析并保存地理位置统计数据
            if (StrUtil.isNotBlank(infoCode) && StrUtil.equals(infoCode, "10000")) {
                String province = localeResultObj.getString("province");
                boolean unknownFlag = StrUtil.equals(province, "[]");  // 判断是否为未知地区

                // 构建地理位置统计对象
                LinkLocaleStatsDO linkLocaleStatsDO = LinkLocaleStatsDO.builder()
                        .province(actualProvince = unknownFlag ? actualProvince : province)  // 省份信息
                        .city(actualCity = unknownFlag ? actualCity : localeResultObj.getString("city"))  // 城市信息
                        .adcode(unknownFlag ? "未知" : localeResultObj.getString("adcode"))  // 行政区划代码
                        .cnt(1)  // 访问次数+1
                        .fullShortUrl(fullShortUrl)
                        .country("中国")  // 默认国家
                        .date(new Date())
                        .build();
                // 保存地理位置统计数据
                linkLocaleStatsMapper.linkLocaleStats(linkLocaleStatsDO);
            }

            // 11. 【操作系统统计】记录和统计访问者的操作系统信息
            LinkOsStatsDO linkOsStatsDO = LinkOsStatsDO.builder()
                    .os(statsRecord.getOs())  // 操作系统类型(Windows/Mac/Linux/Android/iOS等)
                    .cnt(1)  // 访问次数+1
                    .fullShortUrl(fullShortUrl)
                    .date(new Date())
                    .build();
            linkOsStatsMapper.linkOsStats(linkOsStatsDO);

            // 12. 【浏览器统计】记录和统计访问者的浏览器信息
            LinkBrowserStatsDO linkBrowserStatsDO = LinkBrowserStatsDO.builder()
                    .browser(statsRecord.getBrowser())  // 浏览器类型(Chrome/Firefox/Safari/Edge等)
                    .cnt(1)  // 访问次数+1
                    .fullShortUrl(fullShortUrl)
                    .date(new Date())
                    .build();
            linkBrowserStatsMapper.linkBrowserStats(linkBrowserStatsDO);

            // 13. 【设备类型统计】记录和统计访问者的设备类型信息
            LinkDeviceStatsDO linkDeviceStatsDO = LinkDeviceStatsDO.builder()
                    .device(statsRecord.getDevice())  // 设备类型(PC/Mobile/Tablet等)
                    .cnt(1)  // 访问次数+1
                    .fullShortUrl(fullShortUrl)
                    .date(new Date())
                    .build();
            linkDeviceStatsMapper.linkDeviceStats(linkDeviceStatsDO);

            // 14. 【网络类型统计】记录和统计访问者的网络连接类型
            LinkNetworkStatsDO linkNetworkStatsDO = LinkNetworkStatsDO.builder()
                    .network(statsRecord.getNetwork())  // 网络类型(WiFi/4G/5G/宽带等)
                    .cnt(1)  // 访问次数+1
                    .fullShortUrl(fullShortUrl)
                    .date(new Date())
                    .build();
            linkNetworkStatsMapper.linkNetworkStats(linkNetworkStatsDO);

            // 15. 【访问日志记录】保存详细的访问日志信息，用于后续的数据分析和审计
            LinkAccessLogsDO linkAccessLogsDO = LinkAccessLogsDO.builder()
                    .user(statsRecord.getUv())  // 用户唯一标识
                    .ip(statsRecord.getRemoteAddr())  // 访问者IP地址
                    .browser(statsRecord.getBrowser())  // 浏览器信息
                    .os(statsRecord.getOs())  // 操作系统信息
                    .network(statsRecord.getNetwork())  // 网络类型
                    .device(statsRecord.getDevice())  // 设备类型
                    .locale(StrUtil.join("-", "中国", actualProvince, actualCity))  // 完整地理位置
                    .fullShortUrl(fullShortUrl)
                    .build();
            linkAccessLogsMapper.insert(linkAccessLogsDO);

            // 16. 【短链接基础统计更新】更新短链接主表中的PV、UV、UIP总计数
            shortLinkMapper.incrementStats(gid, fullShortUrl, 1,
                    statsRecord.getUvFirstFlag() ? 1 : 0,  // UV增量
                    statsRecord.getUipFirstFlag() ? 1 : 0);  // UIP增量

            // 17. 【今日统计数据】记录当日的PV、UV、UIP统计，用于今日数据的实时展示
            LinkStatsTodayDO linkStatsTodayDO = LinkStatsTodayDO.builder()
                    .todayPv(1)  // 今日PV+1
                    .todayUv(statsRecord.getUvFirstFlag() ? 1 : 0)  // 今日UV(如果是首次访问用户)
                    .todayUip(statsRecord.getUipFirstFlag() ? 1 : 0)  // 今日UIP(如果是首次访问IP)
                    .fullShortUrl(fullShortUrl)
                    .date(new Date())
                    .build();
            linkStatsTodayMapper.shortLinkTodayStats(linkStatsTodayDO);

        } catch (Throwable ex) {
            // 18. 【异常处理】记录统计过程中的任何异常，确保系统稳定性
            log.error("短链接访问量统计异常", ex);
        } finally {
            // 19. 【锁释放】确保分布式锁被正确释放，避免死锁情况
            rLock.unlock();
        }
    }
}
